package com.zim.access.infrastructurre.protocol;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.zim.access.infrastructurre.Constants;
import com.zim.access.infrastructurre.common.AccessConfig;
import com.zim.access.infrastructurre.etcd.EtcdClientAdvanced;
import com.zim.access.infrastructurre.protocol.api.Protocol;
import com.zim.access.infrastructurre.protocol.api.Provider;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import lombok.var;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import java.net.URI;

@Slf4j
public class ProtocolBootstrap implements InitializingBean {

    private static final EventLoopGroup BOSS = new NioEventLoopGroup(1, new ThreadFactoryBuilder().setNameFormat("access-boss-pool-%s").build());
    private static final EventLoopGroup WORKER = new NioEventLoopGroup(2 * Runtime.getRuntime().availableProcessors(), new ThreadFactoryBuilder().setNameFormat("access-worker-pool-%s").build());

    @Autowired
    private AccessConfig config;

    @Autowired
    private Protocol protocol;

    @Autowired
    private EtcdClientAdvanced etcdClientAdvanced;

    @Override
    public void afterPropertiesSet() throws Exception {
        for (String url : config.getBinds()) {
            doStart(url);
        }
    }

    void doStart(String url) {
        URI uri = URI.create(url);
        Provider provider = protocol.ref(uri);

        var server = new ServerBootstrap().
                group(BOSS, WORKER)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.config().setAutoRead(false);

                        ch.pipeline().addLast(provider.getEncoder());

                        ch.pipeline().addLast(provider.handler());
                    }
                });

        server.bind(uri.getHost(), uri.getPort()).addListener(future -> {
            if (future.isSuccess()) {
                if (log.isInfoEnabled()) {
                    log.info("[access] start service {} success", uri);
                }

                // 为当前服务绑定一个租约，进行健康检查
                etcdClientAdvanced.leaseClient().keepAlive(Constants.HEALTH_PREFIX + url, String.valueOf(System.currentTimeMillis()));
            } else {
                if (log.isErrorEnabled()) {
                    log.error("[access] start service {} fail", uri, future.cause());
                }
            }
        });
    }
}
